package org.quincy.rock.comm.process;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import org.quincy.rock.comm.CommunicateException;
import org.quincy.rock.core.exception.BlcokingException;
import org.quincy.rock.core.util.CoreUtil;
import org.quincy.rock.core.util.DateUtil;

/**
 * <b>使用队列缓存的报文处理服务基类。</b>
 * <p><b>详细说明：</b></p>
 * <!-- 在此添加详细说明 -->
 * 无。
 * <p><b>修改列表：</b></p>
 * <table width="100%" cellSpacing=1 cellPadding=3 border=1>
 * <tr bgcolor="#CCCCFF"><td>序号</td><td>作者</td><td>修改日期</td><td>修改内容</td></tr>
 * <!-- 在此添加修改列表，参考第一行内容 -->
 * <tr><td>1</td><td>wks</td><td>2018年5月9日 下午3:17:33</td><td>建立类型</td></tr>
 * 
 * </table>
 * @version 1.0
 * @author wks
 * @since 1.0
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
public abstract class QueueMessageProcessService extends MessageProcessService {
	/**
	 * 阻塞超时秒数。
	 */
	private int timeout = 10;
	/**
	 * 线程组。
	 */
	private final ThreadGroup threadGroup;
	/**
	 * 线程数。
	 */
	private final int threadCount;
	/**
	 * 执行者线程组。
	 */
	private Executor<QueueMessage>[] executors;

	/**
	 * <b>构造方法。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param tc 线程数
	 */
	public QueueMessageProcessService(int tc) {
		this.threadCount = Math.max(tc, 1);
		threadGroup = new ThreadGroup(getId());
		threadGroup.setMaxPriority(Thread.MAX_PRIORITY);
	}

	/**
	 * <b>初始化执行者线程。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 */
	private synchronized void initExecutorThread() {
		if (executors == null) {
			executors = new Executor[threadCount];
			for (int i = 0; i < threadCount; i++) {
				executors[i] = new Executor(i, createBlockingQueue(), threadGroup, getId() + "_" + i);
				executors[i].setPriority(Thread.MAX_PRIORITY);
				executors[i].start();
			}
		}
	}

	/** 
	 * start。
	 * @see org.quincy.rock.comm.process.MessageProcessService#start()
	 */
	@Override
	public void start() {
		if (executors == null)
			initExecutorThread();
		super.start();
	}

	/**
	 * <b>创建阻塞队列。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 阻塞队列
	 */
	protected abstract BlockingQueue<QueueMessage> createBlockingQueue();

	/**
	 * <b>返回当前运行线程个数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 当前运行线程个数
	 */
	public final int activeCount() {
		return threadGroup.activeCount();
	}

	/**
	 * <b>获得线程数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 线程数
	 */
	public final int getThreadCount() {
		return this.threadCount;
	}

	/**
	 * 返回阻塞超时秒数
	 * @return 阻塞超时秒数
	 */
	public final int getTimeout() {
		return timeout;
	}

	/**
	 * 设置阻塞超时秒数
	 * @param timeout 阻塞超时秒数
	 */
	public final void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	/**
	 * <b>返回队列是否为空。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 返回队列是否为空
	 */
	public final boolean isEmpty() {
		if (executors == null)
			return true;
		int len = getThreadCount();
		for (int i = 0; i < len; i++) {
			if (count(i) > 0)
				return false;
		}
		return true;
	}

	/**
	 * <b>返回所有线程是否是空闲的。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 所有线程是否是空闲的
	 */
	public final boolean isIdle() {
		if (executors != null) {
			for (int i = 0, len = getThreadCount(); i < len; i++) {
				if (!isIdle(i))
					return false;
			}
		}
		return true;
	}

	/**
	 * <b>返回指定线程是否是空闲的。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadIndex 线程索引
	 * @return 指定线程是否是空闲的
	 */
	public final boolean isIdle(int threadIndex) {
		if (executors == null)
			return true;
		Executor<QueueMessage> executor = executors[threadIndex];
		return executor == null ? true : executor.isIdle();
	}

	/**
	 * <b>等待队列中所有报文处理完毕。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 */
	public final void waitAllDone() {
		while (!(isEmpty() && isIdle())) {
			DateUtil.sleep(500);
		}
	}

	/**
	 * <b>返回指定线程的待处理报文条数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadindex 线程索引
	 * @return 队列中待处理报文条数
	 */
	public final int count(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<QueueMessage> executor = executors[threadIndex];
		return (executor == null) ? 0 : executor.count();
	}

	/**
	 * <b>返回待处理总条数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 待处理总条数
	 */
	public final int count() {
		if (executors == null)
			return 0;
		int len = getThreadCount();
		int total = 0;
		for (int i = 0; i < len; i++) {
			total += count(i);
		}
		return total;
	}

	/**
	 * <b>返回指定线程的报文等待秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 指示报文在队列中等待多久会被处理,反应线程处理报文的效率。
	 * @param threadIndex 线程索引
	 * @return 指定线程的报文等待秒数
	 */
	public final int waitSeconds(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<QueueMessage> executor = executors[threadIndex];
		return (executor == null) ? 0 : executor.waitSeconds();
	}

	/**
	 * <b>返回报文平均等待秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 指示报文在队列中等待多久会被处理,反映线程的处理效率。
	 * @return 报文平均等待秒数
	 */
	public final int waitSeconds() {
		int len = getThreadCount();
		int total = 0;
		for (int i = 0; i < len; i++) {
			total += waitSeconds(i);
		}
		return total / len;
	}

	/**
	 * <b>>返回指定线程的处理报文条数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadIndex 线程索引
	 * @return 指定线程的处理报文条数
	 */
	public final long processCount(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<QueueMessage> executor = executors[threadIndex];
		return (executor == null) ? 0 : executor.processCount();
	}

	/**
	 * <b>返回处理数据总条数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 处理数据总条数
	 */
	public final long processCount() {
		int len = getThreadCount();
		long total = 0;
		for (int i = 0; i < len; i++) {
			total += processCount(i);
		}
		return total;
	}

	/**
	 * <b>>返回指定线程的单个报文处理毫秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadIndex 线程索引
	 * @return 指定线程的单个报文处理毫秒数
	 */
	public final long processMillis(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<QueueMessage> executor = executors[threadIndex];
		return (executor == null) ? 0 : executor.processMillis();
	}

	/**
	 * <b>返回单条报文平均处理毫秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 单条报文平均处理毫秒数
	 */
	public final long processMillis() {
		int len = getThreadCount();
		long total = 0;
		for (int i = 0; i < len; i++) {
			total += processMillis(i);
		}
		return total / len;
	}

	/**
	 * <b>>返回指定线程的单个报文解析毫秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadIndex 线程索引
	 * @return 指定线程的单个报文解析毫秒数
	 */
	public final long resolveMillis(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<QueueMessage> executor = executors[threadIndex];
		return (executor == null) ? 0 : executor.resolveMillis();
	}

	/**
	 * <b>返回单个报文解析平均毫秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 单个报文解析平均毫秒数
	 */
	public final long resolveMillis() {
		int len = getThreadCount();
		long total = 0;
		for (int i = 0; i < len; i++) {
			total += resolveMillis(i);
		}
		return total / len;
	}

	/**
	 * <b>>返回指定线程的单个报文传输毫秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadIndex 线程索引
	 * @return 指定线程的单个报文传输毫秒数
	 */
	public final long transferMillis(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<QueueMessage> executor = executors[threadIndex];
		return (executor == null) ? 0 : executor.transferMillis();
	}

	/**
	 * <b>返回单个报文传输平均毫秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 单个报文传输平均毫秒数
	 */
	public final long transferMillis() {
		int len = getThreadCount();
		long total = 0;
		for (int i = 0; i < len; i++) {
			total += transferMillis(i);
		}
		return total / len;
	}

	/** 
	 * handleArrivedMessage。
	 * @see org.quincy.rock.comm.process.MessageProcessService#handleArrivedMessage(org.quincy.rock.comm.process.QueueMessage)
	 */
	@Override
	protected final void handleArrivedMessage(QueueMessage queueMessage) {
		int threadIndex = assignExecutorIndex(queueMessage.terminalId(), getThreadCount());
		executors[threadIndex].put(queueMessage);
	}

	/**
	 * <b>获得最轻松的线程索引。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param totalCount 总共线程数
	 * @return 最轻松的线程索引
	 */
	protected final int easiest(int totalCount) {
		int minCount = count(0), minIndex = 0;
		if (minCount != 0) {
			for (int index = 1; index < totalCount; index++) {
				int count = count(index);
				if (count == 0) {
					minIndex = index;
					break;
				} else if (minCount > count) {
					minCount = count;
					minIndex = index;
				}
			}
		}
		return minIndex;
	}

	/**
	 * <b>分配线程执行者索引。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param terminalId 终端标识
	 * @param totalCount 总共线程数
	 * @return 线程执行者索引
	 */
	protected int assignExecutorIndex(Object terminalId, int totalCount) {
		int v = (terminalId instanceof Number) ? ((Number) terminalId).intValue() : terminalId.toString().hashCode();
		return Math.abs(v) % totalCount;
	}

	/** 
	 * awaitTermination。
	 * @see org.quincy.rock.comm.process.MessageProcessService#awaitTermination(long)
	 */
	@Override
	protected void awaitTermination(long timeout) throws Exception {
		if (executors == null)
			return;
		int len = executors.length;
		//设置停止标记
		for (int i = 0; i < len; i++) {
			Executor exec = executors[i];
			if (exec != null)
				executors[i].setStopFlag();
		}
		//join
		if (timeout > 0) {
			timeout *= 1000;
			long base = System.currentTimeMillis();
			for (int i = 0; i < len; i++) {
				Executor exec = executors[i];
				if (exec != null) {
					exec.join(timeout);
					timeout = timeout - (System.currentTimeMillis() - base);
					if (timeout > 0)
						base = System.currentTimeMillis();
					else
						break; //没有剩余超时了。
				}
			}
		} else {
			for (int i = 0; i < len; i++) {
				Executor exec = executors[i];
				if (exec != null)
					exec.join(timeout);
			}
		}
		//置空
		for (int i = 0; i < len; i++) {
			CoreUtil.interrupt(executors[i]);
			executors[i] = null;
		}
		executors = null;
		CoreUtil.interrupt(threadGroup);
		threadGroup.destroy();
	}

	//执行者
	private class Executor<T extends QueueMessage> extends Thread {
		/**
		 * threadIndex。
		 */
		private final int threadIndex;
		/**
		 * queue。
		 */
		private BlockingQueue<T> queue;
		/**
		 * 线程停止标记。
		 */
		private boolean stopFlag;
		/**
		 * 报文传输毫秒数。
		 */
		private long transferMillis;
		/**
		 * 报文解析毫秒数。
		 */
		private long resolveMillis;
		/**
		 * 单个报文处理毫秒数。
		 */
		private long processMillis;
		/**
		 * 处理报文条数。
		 */
		private long processCount;
		/**
		 * 报文等待秒数。
		 */
		private int waitSeconds;
		/**
		 * 空闲。
		 */
		private boolean idle;
		/**
		 * 最近一次工作时间。
		 */
		private long lastWorkTime;

		/**
		 * <b>构造方法。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @param index
		 * @param queue 队列
		 * @param group 线程组
		 * @param name 线程名称
		 */
		public Executor(int index, BlockingQueue<T> queue, ThreadGroup group, String name) {
			super(group, name);
			this.threadIndex = index;
			this.queue = queue;
		}

		/**
		 * <b>返回队列。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 队列
		 */
		private BlockingQueue<T> queue() {
			return queue;
		}

		/** 
		 * run。
		 * @see java.lang.Thread#run()
		 */
		@Override
		public void run() {
			while (true) {
				try {
					this.idle = true;
					T msg = queue().poll(500, TimeUnit.MILLISECONDS);
					if (msg != null) {
						this.idle = false;
						recorder.write("Retrieve the message from the queue:{0},term:{1}.", threadIndex,
								msg.terminalId());
						long ibegin = System.currentTimeMillis();
						this.waitSeconds = (int) (ibegin - msg.timestamp()) / 1000;
						processQueueMessage(msg);
						long iEnd = System.currentTimeMillis();
						this.processMillis = iEnd - ibegin;
						this.processCount++;
						this.lastWorkTime = iEnd;
						this.idle = true;
					}
					if (stopFlag && queue.isEmpty()) {
						//退出并销毁线程			
						recorder.write("Quit the queue:{0}.", threadIndex);
						break;
					}
				} catch (Exception e) {
					recorder.write(e, e.getMessage());
				}
			}
			this.idle = true;
			Executor[] execs = executors;
			if (execs != null)
				execs[threadIndex] = null;
		}

		/**
		 * <b>放入一条队列报文。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @param msg 队列报文
		 */
		public void put(T msg) {
			try {
				this.resolveMillis = System.currentTimeMillis() - msg.timestamp(); //报文解析时间
				Long messageTime = msg.messageTime();
				if (messageTime != null)
					this.transferMillis = msg.timestamp() - messageTime; //报文传输时间
				if (stopFlag || timeout > 0 ? !queue().offer(msg, timeout, TimeUnit.SECONDS) : !queue().offer(msg)) {
					BlcokingException ex = new BlcokingException("terminalId:" + msg.terminalId());
					recorder.write(ex, ex.getMessage());
					throw ex;
				} else if (recorder.canWrite()) {
					recorder.write("The message has been put on the queue:{0},term:{1}.", threadIndex,
							msg.terminalId());
				}
			} catch (InterruptedException e) {
				throw new CommunicateException(e.getMessage(), e);
			}
		}

		/**
		 * <b>返回队列中待处理报文个数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 队列中待处理报文个数
		 */
		public int count() {
			return queue().size();
		}

		/**
		 * <b>线程是空闲的。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 线程是空闲的
		 */
		public boolean isIdle() {
			return idle && (System.currentTimeMillis() - lastWorkTime) > 1500;
		}

		/**
		 * <b>报文等待秒数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 指示报文在队列中等待多久会被处理。
		 * @return 报文等待秒数
		 */
		public int waitSeconds() {
			return this.waitSeconds;
		}

		/**
		 * <b>处理报文条数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 处理报文条数
		 */
		public long processCount() {
			return this.processCount;
		}

		/**
		 * <b>单个报文处理毫秒数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 单个报文处理毫秒数
		 */
		public long processMillis() {
			return this.processMillis;
		}

		/**
		 * <b>报文解析毫秒数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 报文解析毫秒数
		 */
		public long resolveMillis() {
			return this.resolveMillis;
		}

		/**
		 * <b>报文传输毫秒数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 报文传输毫秒数
		 */
		public long transferMillis() {
			return this.transferMillis;
		}

		/**
		 * <b>设置停止标志。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 通知停止执行者线程。
		 */
		public void setStopFlag() {
			stopFlag = true;
		}
	}
}
